package com.yy.smartBI.mqtest;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.DeliverCallback;

import java.nio.charset.StandardCharsets;

public class DieLetterWorker {
    private final static String QUEUE_NAME1 = "m3";
    private final static String QUEUE_NAME2 = "m4";


    public static void main(String[] argv) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        Connection connection = factory.newConnection();
        Channel channel1 = connection.createChannel();
        Channel channel2 = connection.createChannel();
        //声明死信交换机
        channel1.exchangeDeclare("e2", "direct");

        //声明死信队列
        channel1.queueDeclare(QUEUE_NAME1, true, false, false, null);
        //绑定队列到交换机
        channel1.queueBind(QUEUE_NAME1, "e2", "worker1");

        //声明交换机 同上
        channel2.exchangeDeclare("e2", "direct");
        channel2.queueDeclare(QUEUE_NAME2, true, false, false, null);
        channel2.queueBind(QUEUE_NAME2, "e2", "worker2");


        System.out.println(" [*] Waiting for messages. To exit press CTRL+C");

        DeliverCallback deliverCallback1 = (consumerTag, delivery) -> {
            try {
                String message = new String(delivery.getBody(), StandardCharsets.UTF_8);
                System.out.println(" [x]worker1 Received1 '" + message + "'");
                //模拟消息处理
                Thread.sleep(1000);
                //如果消息处理失败，将消息发送到死信队列
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        };

        DeliverCallback deliverCallback2 = (consumerTag, delivery) -> {
            try {
                String message = new String(delivery.getBody(), StandardCharsets.UTF_8);
                System.out.println(" [x]worker2 Received1 '" + message + "'");
                //模拟消息处理
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        };

        channel1.basicConsume(QUEUE_NAME1, true, deliverCallback1, consumerTag -> {
        });
        channel2.basicConsume(QUEUE_NAME2, true, deliverCallback2, consumerTag -> {
        });
    }
}
